package org.example.kafka.sendtype;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.example.kafka.common.CommonConstant;

import java.util.Collections;
import java.util.Properties;

/**
 * kafka消息的消费者
 * @Author:anlu
 */
public class KafkaConsumer1 {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers","127.0.0.1:9092");
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer", StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"consumer-test2");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);
        try{
            // 消费者订阅这个主题
            consumer.subscribe(Collections.singletonList(CommonConstant.HELLO_TOPIC));
            // 消费者主动去拉取消息
            while (true){
                ConsumerRecords<String,String> records = consumer.poll(500);
                for(ConsumerRecord<String, String> record:records){
                    System.out.println(String.format("topic:%s,分区：%d,偏移量：%d," +
                                    "key:%s,value:%s",record.topic(),record.partition(),
                            record.offset(),record.key(),record.value()));
                    //do my work

                }
            }
        }finally {
           consumer.close();
        }

    }
}
